// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: fengyang

#include <limits>

#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed/timer/timer_manager.h"

namespace coin2::exchange::base::feed::timer {
namespace impl {

std::shared_ptr<TimerManager> TimerManager::instance_;

TimerManager::TimerManager() {
}

TimerManager::~TimerManager() {
}

uint64_t TimerManager::NewTimeId() {
  timer_id_++;
  return timer_id_;
}

void TimerManager::CheckExpires(int64_t now) {
  std::pair<int64_t, uint64_t> now_entry{now, std::numeric_limits<uint64_t>::max()};
  auto last = timers_.upper_bound(now_entry);
  auto expired = timers_.begin();
  std::vector<std::tuple<int64_t, uint64_t, TimerProperty>> expired_timers;
  for ( ; expired != last; expired++) {
    expired_timers.push_back(std::make_tuple(
      expired->first, expired->second, timer_property_[expired->second]));
  }
  timers_.erase(timers_.begin(), last);

  for (auto& timer : expired_timers) {
    int64_t timestamp;
    uint64_t timer_id;
    TimerProperty property;
    std::tie(timestamp, timer_id, property) = timer;
    if (property.periodic) {
      int64_t next_timestamp = timestamp + property.duration;
      timers_.insert(std::make_pair(next_timestamp, timer_id));
    } else {
      timer_property_.erase(timer_id);
    }
  }

  // do callback is the last action as user may cancel timer in callback
  for (auto& timer : expired_timers) {
    int64_t timestamp;
    uint64_t timer_id;
    TimerProperty property;
    std::tie(timestamp, timer_id, property) = timer;
    feed_timer_.SetTimerId(timer_id);
    feed_timer_.SetTimestamp(timestamp);
    property.callback();
  }
}

void TimerManager::AddTimer(int64_t now, uint64_t id, const TimerProperty& property) {
  if (property.duration > 0) {
    int64_t expires_timestamp = now + property.duration;
    timers_.insert({expires_timestamp, id});
  } else {
    timers_.insert({property.timestamp, id});
  }

  timer_property_[id] = property;
}

int64_t TimerManager::Expires(uint64_t id) {
  for (auto& item : timers_) {
    if (item.second == id) {
      return item.first;
    }
  }

  return 0;
}

void TimerManager::RemoveTimer(uint64_t id) {
  for (auto& item : timers_) {
    if (item.second == id) {
      timers_.erase(item);
      break;
    }
  }
  timer_property_.erase(id);
}

void TimerManager::TriggerTimer() {
  CHECK(subscriber_);
  subscriber_->onHeartbeatFeed(feed_timer_);
}

std::shared_ptr<ArchiveTimerManager> TimerManager::CreateArchiveManager() {
  CHECK(!instance_);
  std::shared_ptr<ArchiveTimerManager> ptr(new ArchiveTimerManager());
  instance_ = ptr;
  return ptr;
}

std::shared_ptr<LiveTimerManager> TimerManager::CreateLiveManager() {
  CHECK(!instance_);
  std::shared_ptr<LiveTimerManager> ptr(new LiveTimerManager());
  instance_ = ptr;
  return ptr;
}

std::shared_ptr<TimerManager> TimerManager::GetInstance() {
  CHECK(instance_);
  return instance_;
}

void TimerManager::DeleteInstance() {
  if (instance_) {
    instance_.reset();
  }
}

ArchiveTimerManager::ArchiveTimerManager() {
}

ArchiveTimerManager::~ArchiveTimerManager() {
}

int64_t ArchiveTimerManager::ExpiresAt(uint64_t id, int64_t timestamp, std::function<void(void)> callback) {
  TimerProperty property{false, 0, timestamp, callback};
  if (now_) {
    AddTimer(now_.value(), id, property);
  } else {
    pending_timers.insert(std::make_pair(id, property));
  }

  return id;
}

int64_t ArchiveTimerManager::ExpiresFromNow(uint64_t id, int64_t duration, std::function<void(void)> callback) {
  TimerProperty property{false, duration, 0, callback};
  if (now_) {
    AddTimer(now_.value(), id, property);
  } else {
    pending_timers.insert(std::make_pair(id, property));
  }

  return id;
}

int64_t ArchiveTimerManager::ExpiresEvery(uint64_t id, int64_t duration, std::function<void(void)> callback) {
  TimerProperty property{true, duration, 0, callback};
  if (now_) {
    AddTimer(now_.value(), id, property);
  } else {
    pending_timers.insert(std::make_pair(id, property));
  }

  return id;
}

void ArchiveTimerManager::Cancel(uint64_t id) {
  RemoveTimer(id);
  pending_timers.erase(id);
}

void ArchiveTimerManager::Tick(int64_t now) {
  if (!now_) {
    for (auto item : pending_timers) {
      AddTimer(now, item.first, item.second);
    }
  }

  // archive feed may be spare, timer may be expired several times
  while (!timers_.empty() && timers_.begin()->first <= now){
    now_ = timers_.begin()->first;
    CheckExpires(now_.value());
  }

  now_ = now;
}

int64_t ArchiveTimerManager::Expires(uint64_t id) {
  if (!now_) {
    if (pending_timers.count(id) == 0) {
      return 0;
    }

    if (pending_timers.at(id).duration > 0) {
      return pending_timers.at(id).duration;
    }

    return pending_timers.at(id).timestamp;
  }

  return TimerManager::Expires(id);
}

IFeedSubscriber* ArchiveTimerManager::HookSubscriber(IFeedSubscriber* subscriber) {
  subscriber_ = subscriber;
  timer_subscriber_ = std::make_unique<TimerFeedSubscriber>(subscriber, this);
  return timer_subscriber_.get();
}


LiveTimerManager::LiveTimerManager() {
}

LiveTimerManager::~LiveTimerManager() {
}

int64_t LiveTimerManager::ExpiresAt(uint64_t id, int64_t timestamp, std::function<void(void)> callback) {
  TimerProperty property{false, 0, timestamp, callback};
  AddTimer(GetCurrentTimestamp(), id, property);
  ResetTimerExpiry();
  return id;
}

int64_t LiveTimerManager::ExpiresFromNow(uint64_t id, int64_t duration, std::function<void(void)> callback) {
  TimerProperty property{false, duration, 0, callback};
  AddTimer(GetCurrentTimestamp(), id, property);
  ResetTimerExpiry();
  return id;
}

int64_t LiveTimerManager::ExpiresEvery(uint64_t id, int64_t duration, std::function<void(void)> callback) {
  TimerProperty property{true, duration, 0, callback};
  AddTimer(GetCurrentTimestamp(), id, property);
  ResetTimerExpiry();
  return id;
}

void LiveTimerManager::Cancel(uint64_t id) {
  RemoveTimer(id);
  ResetTimerExpiry();
}

void LiveTimerManager::Tick(int64_t now) {
  while (!timers_.empty() && timers_.begin()->first <= now){
    CheckExpires(now);
  }
}

void LiveTimerManager::RegisterSubscriber(IFeedSubscriber* subscriber) {
  subscriber_ = subscriber;
}

void LiveTimerManager::RegisterTimerIoc(boost::asio::io_context& ioc) {
  tick_timer_ = std::make_unique<boost::asio::deadline_timer>(ioc);
}

void LiveTimerManager::ResetTimerExpiry() {
  if (timers_.size() > 0) {
    auto first_expired = timers_.begin();
    tick_timer_->expires_at(PtimeFromTimestamp(first_expired->first));
    tick_timer_->async_wait(std::bind(
        &LiveTimerManager::TimerCallback, this));
  } else if (tick_timer_) {
    tick_timer_->cancel();
  } else {
    NOTREACHED() << "Timer not initialized.";
  }
}

void LiveTimerManager::TimerCallback() {
  auto now = GetCurrentTimestamp();
  Tick(now);
  ResetTimerExpiry();
}

}  // namespace impl

}  // namespace coin2::exchange::base::feed::timer
